package com.zhangx.study.demo.rxjava2;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.slf4j.Slf4j;

/**
 * @ClassName HelloRxjava2
 * @Description rxjava主要用在安卓
 * https://juejin.cn/post/6844903617124630535
 * @Author zhangx
 * @Date 2021/5/26 8:33
 **/
@Slf4j
public class HelloRxjava2 {

    public static void main(String[] args) {
//        一个可观察的对象，泛型就是事件对象
        Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> observableEmitter) throws Exception {
                observableEmitter.onNext("i am sleeping");
                observableEmitter.onNext("i am working");
                log.info("可被观察对象发送了事件");
//                compelete后 不会继续发送了
                observableEmitter.onComplete();
                observableEmitter.onNext("=============");
            }
        });

//        消费者
        Consumer<String> consumer = new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                log.info("接收到事件={}",s);
            }
        };

//        观察者，比consumer 更复杂一点
        Observer<String> observer = new Observer<String>() {
            @Override
            public void onSubscribe(Disposable disposable) {
                log.info("observer onSubscribe disposable.isDisposed()={}",disposable.isDisposed());
            }

            @Override
            public void onNext(String s) {
                log.info("observer 接收到事件={}",s);
            }

            @Override
            public void onError(Throwable throwable) {

            }

            @Override
            public void onComplete() {
                log.info("observer onComplete");
            }
        };
//      一个可被观察的对象，被订阅了消费者--典型被动语法
//        observable.subscribe(consumer);
//        开启新线程消费
//        observable.observeOn(Schedulers.single()).subscribe(consumer);
        observable.observeOn(Schedulers.single()).subscribe(observer);
        try {
            Thread.sleep(2000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
